package com.atguigu.bigdata.gmall.realtime.app.dwd;

import com.atguigu.bigdata.gmall.realtime.app.BaseSQLApp;
import com.atguigu.bigdata.gmall.realtime.common.Constant;
import com.atguigu.bigdata.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/10/10 14:26
 */
public class Dwd_02_DwdTradeCartAdd extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_02_DwdTradeCartAdd().init(
            3002,
            2,
            "Dwd_02_DwdTradeCartAdd"
        );
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        // 1. 消费 ods_db
        // ddl方式
        readOdsDb(tEnv, "Dwd_02_DwdTradeCartAdd");
        // sql中的 print 是阻塞式方法
        //        tEnv.sqlQuery("select * from ods_db").execute().print();
        // 2. 从 ods_db中过滤出加购
        Table cartInfo = tEnv.sqlQuery("select " +
                                           " `data`['id'] id, " +
                                           " `data`['user_id'] user_id, " +
                                           " `data`['sku_id'] sku_id, " +
                                           " `data`['source_id'] source_id, " +
                                           " `data`['source_type'] source_type," +
                                           " if(`type`='insert', " +
                                           "   `data`['sku_num']," +
                                           "    cast(cast(`data`['sku_num'] as int) - cast(`old`['sku_num'] as int) as string)" +
                                           " ) sku_num, " +
                                           " ts," +
                                           " pt " +
                                           "from ods_db " +
                                           "where `database`='gmall2022' " +
                                           "and `table`='cart_info' " +
                                           "and (" +
                                           " `type`='insert' " +
                                           "  or " +
                                           "   (`type`='update' " +
                                           "     and `old`['sku_num'] is not null " +
                                           "     and cast(`data`['sku_num'] as int) > cast(`old`['sku_num']  as int)" +
                                           "   )" +
                                           ")");
        tEnv.createTemporaryView("cart_info", cartInfo);
        
        // 3. 读字典表
        readBaseDic(tEnv);
        
        // 4. 加购和字典表进行 lookup join
        Table result = tEnv.sqlQuery("select " +
                                         "ci.id,  " +
                                         "user_id,  " +
                                         "sku_id,  " +
                                         "source_id,  " +
                                         "source_type,  " +
                                         "dic_name source_type_name,  " +
                                         "sku_num,  " +
                                         "ts  " +
                                         "from cart_info ci " +
                                         "join base_dic for system_time as of ci.pt as dic " +
                                         "on ci.source_type=dic.dic_code");
        
        // 5. 写出到 kafka 中
        tEnv.executeSql("create table dwd_trade_cart_add(" +
                            "id string,  " +
                            "user_id string,  " +
                            "sku_id string,  " +
                            "source_id string,  " +
                            "source_type_code string,  " +
                            "source_type_name string,  " +
                            "sku_num string,  " +
                            "ts bigint  " +
                            ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_TRADE_CART_ADD));
        
        result.executeInsert("dwd_trade_cart_add");
    }
}
/*
加购的过滤:
    database = gmall2022
    table = cart_info
    type= insert update
    
    sku_1      2     新增
    sku_1      5-2   update
        insert 或者 update 的时候,sku_num 变大
 */